package com.hopes.bizz.protocol.hj212.server.handler;

import cn.allbs.hj212.format.T212Mapper;
import cn.allbs.hj212.model.HjData;
import cn.hutool.cache.Cache;
import cn.hutool.cache.CacheUtil;
import cn.hutool.core.lang.Pair;
import cn.hutool.crypto.digest.DigestUtil;
import com.hopes.bizz.common.core.util.StringUtils;
import com.hopes.bizz.protocol.hj212.bo.ClientChannelRel;
import com.hopes.bizz.protocol.hj212.bo.HjDataBuild;
import com.hopes.bizz.protocol.hj212.bo.HjDataParse;
import com.hopes.bizz.protocol.hj212.constant.HJ212Constant;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

@Component
@ChannelHandler.Sharable
@Slf4j
public class HJ212InboundHandler extends SimpleChannelInboundHandler<String> {
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;
    @Autowired
    private DataHandlerChain dataHandlerChain;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public static final String PATTERN_212 = "^##\\d{4}.+&&.{4}(\r\n)*$";
    public static final Pattern mnPattern = Pattern.compile("mn=(.*?);", Pattern.CASE_INSENSITIVE);
    public static final Pattern qnPattern = Pattern.compile("qn=(.*?);", Pattern.CASE_INSENSITIVE);
    public static ChannelGroup recipients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private static final String DUPLICATE_KEY = "TRANSFER:DUPLICATE:";
    private static final Cache<String, List<HjData>> cache = CacheUtil.newTimedCache(1000 * 60 * 30);

    @Override
    protected synchronized void channelRead0(ChannelHandlerContext ctx, String data) {
        log.info("接收报文：{}", data);
        data = data.trim();
        // 是否满足212格式
        if (!data.matches(PATTERN_212)) {
            log.error("数据不满足HJ212格式要求. data={}", data);
        }
        //去重
        if (isRepeatData(data)) {
            log.info("重复的包数据,不处理! data={}", data);
            return;
        }
        //设置客户端与通讯通道对应关系
        setClientChannelRel(data, ctx);
        // 判断是否有分包标识
        Pair<Boolean, HjDataParse> pair = this.isUpPacket(data);
        if (pair.getKey()) {
            HjDataParse hjDataParse = pair.getValue();
            String key = hjDataParse.getKey();
            List<HjData> cacheList = Optional.ofNullable(cache.get(key)).orElseGet(ArrayList::new);
            cacheList.add(hjDataParse.getHjData());
            // 判断是否需要合并
            if (isMerge(cacheList)) {
                // 合并
                HjData mergeData = HjDataBuild.mergeHjData(cacheList);
                String mergeDataStr = HjDataBuild.writeHjDataAsString(mergeData);
                log.info("HJ212数据合并数据: {}", mergeDataStr);
                // 解析
                parseData(mergeDataStr);
                // 移除缓存
                log.info("移除缓存key:{}", key);
                cache.remove(key);
            } else {
                log.info("新增缓存key:{}", key);
                cache.put(key, cacheList);
            }
        } else {
            // 解析数据
            parseData(data);
        }
    }

    private void parseData(String data) {
        //执行数据解析
        taskExecutor.execute(new DataParseThread(data, dataHandlerChain));
    }

    // 判断是否需要合并包
    private Boolean isMerge(List<HjData> list) {
        // 至少两个包合并
        if (list.size() == 1) {
            return false;
        }

        // 排序
        list.sort(Comparator.comparingInt(HjData::getPNo));

        // 判断是否每个序列包都存在
        int index = 1;
        int pNum = 1;
        for (HjData hjData : list) {
            int pNo = hjData.getPNo();
            pNum = hjData.getPNum();
            if (pNo == pNum) {
                break;
            }
            if (pNo == index) {
                index++;
            } else {
                break;
            }
        }

        return index == pNum;
    }

    private boolean isRepeatData(String data) {
        T212Mapper mapper = new T212Mapper().enableDefaultParserFeatures().enableDefaultVerifyFeatures();
        try {
            Map<String, Object> stringObjectMap = mapper.readDeepMap(data);
            String cp = stringObjectMap.get(HJ212Constant.KEY_CP).toString();
            String mn = stringObjectMap.get(HJ212Constant.KEY_MN).toString();
            String cn = stringObjectMap.get(HJ212Constant.KEY_COMMAND_NO).toString();
            String key = String.join(":", cp, mn, cn);
            String md5Key = DUPLICATE_KEY + DigestUtil.md5Hex(key);
            String value = redisTemplate.opsForValue().get(md5Key);
            if (value != null) {
                return true;
            } else {
                redisTemplate.opsForValue().set(md5Key, UUID.randomUUID().toString(), 60 * 60, TimeUnit.SECONDS);
                log.info("新增去重缓存: {}", key);
            }
        } catch (Exception e) {
            log.info("去重失败,格式化错误: {}", data);
        }
        return false;
    }

    private Pair<Boolean, HjDataParse> isUpPacket(String data) {
        HjDataParse hjDataParse = null;
        try {
            hjDataParse = new HjDataParse(data);
        } catch (Exception e) {
            log.info("HJ212数据解析错:{}", data);
        }
        return new Pair<>(Objects.nonNull(hjDataParse) && hjDataParse.isUpPacket(), hjDataParse);
    }

    private void setClientChannelRel(String data, ChannelHandlerContext ctx) {
        String clientId = this.getMnFromHJ212Data(data);
        if (clientId != null) {
            ChannelId channelId = ClientChannelRel.get(clientId);
            if (channelId == null || !channelId.equals(ctx.channel().id())) {
                ClientChannelRel.put(clientId, ctx.channel().id());
            }
        }
    }

    private String getMnFromHJ212Data(String data) {
        Map<String, String> map = AbstractHJ212DataHandler.hj212DataToMap(data);
        String clientId = map.get(HJ212Constant.KEY_MN);
        if (StringUtils.isNotEmpty(clientId)) {
            return clientId;
        }
        return null;
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        String remoteAdderss = channel.remoteAddress().toString();
        channel.closeFuture().addListener(future -> {
            log.debug("通道即将关闭 ==>{}", remoteAdderss);
            ClientChannelRel.removeChannelId(channel.id());
            ClientChannelRel.output();
            if (ctx != null) {
                ctx.close();
            }
        });
        log.info("新的客户端连接：{}", remoteAdderss);
        // 添加到消息交互组
        recipients.add(channel);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.debug("channel inactive, close ctx ==>{}", ctx);
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("有异常==>", cause);
        ctx.close();
    }

    public void writeHJ212DataToClient(String clientId, String hj212Data) {
        log.debug("begin 发送HJ212消息给客户端 ===> clientId: {}, hj212Data:{}", clientId, hj212Data);
        ChannelId channelId = ClientChannelRel.get(clientId);
        if (channelId == null) {
            log.debug("当前没有对应的客户端链接通道，无法发送消息 ===> clientId: {}, hj212Data:{}", clientId, hj212Data);
            return;
        }
        Channel channel = recipients.find(channelId);
        if (channel == null || !channel.isWritable()) {
            log.debug("当前无可用的客户端链接通道，无法发送消息 ===> clientId: {}, hj212Data:{}", clientId, hj212Data);
            return;
        }

        channel.writeAndFlush(hj212Data);
        log.debug("end 发送HJ212消息给客户端 ===> clientId: {}, hj212Data:{}", clientId, hj212Data);
    }

}
